#include "mqtt_client.h"
#include <stdarg.h>
#include <errno.h>

// 全局变量
static mqtt_client_t *g_client = NULL;
static mqtt_message_callback_t g_message_callback = NULL;
static mqtt_connect_callback_t g_connect_callback = NULL;
static mqtt_disconnect_callback_t g_disconnect_callback = NULL;
static volatile int g_running = 1;

// 信号处理函数
void signal_handler(int sig) {
    mqtt_log(2, "Received signal %d, shutting down...\n", sig);
    g_running = 0;
}

// 日志函数
void mqtt_log(int level, const char* format, ...) {
    if (level > MQTT_LOG_LEVEL) return;
    
    va_list args;
    va_start(args, format);
    
    time_t now = time(NULL);
    struct tm *tm_info = localtime(&now);
    
    char time_str[32];
    strftime(time_str, sizeof(time_str), "%Y-%m-%d %H:%M:%S", tm_info);
    
    const char* level_str[] = {"OFF", "ERROR", "WARN", "INFO"};
    printf("[%s][%s] ", time_str, level_str[level]);
    vprintf(format, args);
    
    va_end(args);
}

// mosquitto回调函数
void on_connect(struct mosquitto *mosq, void *obj, int rc) {
    mqtt_client_t *client = (mqtt_client_t*)obj;
    
    mqtt_log(3, "on_connect回调被触发，rc=%d\n", rc);
    
    if (rc == 0) {
        pthread_mutex_lock(&client->mutex);
        client->state = MQTT_STATE_CONNECTED;
        client->reconnect_count = 0;
        client->connect_count++;
        pthread_mutex_unlock(&client->mutex);
        
        mqtt_log(3, "Connected to MQTT broker successfully\n");
        
        // 订阅默认主题
        mqtt_log(3, "订阅主题: %s\n", MQTT_TOPIC_COMMAND);
        int sub_rc = mosquitto_subscribe(mosq, NULL, MQTT_TOPIC_COMMAND, MQTT_QOS_1);
        if (sub_rc != MOSQ_ERR_SUCCESS) {
            mqtt_log(1, "订阅失败: %s\n", mosquitto_strerror(sub_rc));
        }
        
        if (g_connect_callback) {
            g_connect_callback(rc);
        }
    } else {
        pthread_mutex_lock(&client->mutex);
        client->state = MQTT_STATE_ERROR;
        pthread_mutex_unlock(&client->mutex);
        
        mqtt_log(1, "Failed to connect to MQTT broker, return code: %d\n", rc);
        switch(rc) {
            case 1: mqtt_log(1, "   原因: 协议版本不支持\n"); break;
            case 2: mqtt_log(1, "   原因: 客户端ID不合法\n"); break;
            case 3: mqtt_log(1, "   原因: 代理服务器不可用\n"); break;
            case 4: mqtt_log(1, "   原因: 用户名或密码错误\n"); break;
            case 5: mqtt_log(1, "   原因: 未授权\n"); break;
            default: mqtt_log(1, "   原因: 未知错误\n"); break;
        }
    }
}

void on_disconnect(struct mosquitto *mosq, void *obj, int rc) {
    mqtt_client_t *client = (mqtt_client_t*)obj;
    
    pthread_mutex_lock(&client->mutex);
    if (rc == 0) {
        client->state = MQTT_STATE_DISCONNECTED;
        mqtt_log(3, "Disconnected from MQTT broker\n");
    } else {
        client->state = MQTT_STATE_RECONNECTING;
        mqtt_log(2, "Unexpected disconnect from MQTT broker (rc=%d)\n", rc);
    }
    pthread_mutex_unlock(&client->mutex);
    
    if (g_disconnect_callback) {
        g_disconnect_callback(rc);
    }
}

void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg) {
    mqtt_client_t *client = (mqtt_client_t*)obj;
    
    if (msg->payloadlen > 0) {
        // 确保消息以null结尾
        char *payload = malloc(msg->payloadlen + 1);
        if (payload) {
            memcpy(payload, msg->payload, msg->payloadlen);
            payload[msg->payloadlen] = '\0';
            
            pthread_mutex_lock(&client->mutex);
            client->messages_received++;
            pthread_mutex_unlock(&client->mutex);
            
            mqtt_log(3, "Received message on topic '%s': %s\n", msg->topic, payload);
            
            if (g_message_callback) {
                g_message_callback(msg->topic, payload, msg->payloadlen);
            }
            
            free(payload);
        }
    }
}

void on_publish(struct mosquitto *mosq, void *obj, int mid) {
    mqtt_client_t *client = (mqtt_client_t*)obj;
    
    pthread_mutex_lock(&client->mutex);
    client->messages_sent++;
    pthread_mutex_unlock(&client->mutex);
    
    mqtt_log(3, "Message published successfully (mid=%d)\n", mid);
}

void on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos) {
    mqtt_log(3, "Subscribed successfully (mid=%d)\n", mid);
}

void on_unsubscribe(struct mosquitto *mosq, void *obj, int mid) {
    mqtt_log(3, "Unsubscribed successfully (mid=%d)\n", mid);
}

// MQTT客户端初始化
int mqtt_client_init(mqtt_client_t *client, const char* client_id) {
    if (!client || !client_id) {
        mqtt_log(1, "Invalid parameters for mqtt_client_init\n");
        return -1;
    }
    
    memset(client, 0, sizeof(mqtt_client_t));
    
    // 初始化互斥锁
    if (pthread_mutex_init(&client->mutex, NULL) != 0) {
        mqtt_log(1, "Failed to initialize mutex\n");
        return -1;
    }
    
    // 设置默认配置
    strncpy(client->broker_host, MQTT_BROKER_HOST, sizeof(client->broker_host) - 1);
    client->broker_port = MQTT_BROKER_PORT;
    strncpy(client->client_id, client_id, sizeof(client->client_id) - 1);
    client->state = MQTT_STATE_DISCONNECTED;
    
    // 初始化mosquitto库
    int ret = mosquitto_lib_init();
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to initialize mosquitto library: %s\n", mosquitto_strerror(ret));
        pthread_mutex_destroy(&client->mutex);
        return -1;
    }
    
    // 创建mosquitto实例
    client->mosq = mosquitto_new(client->client_id, true, client);
    if (!client->mosq) {
        mqtt_log(1, "Failed to create mosquitto instance\n");
        mosquitto_lib_cleanup();
        pthread_mutex_destroy(&client->mutex);
        return -1;
    }
    
    // 设置回调函数
    mosquitto_connect_callback_set(client->mosq, on_connect);
    mosquitto_disconnect_callback_set(client->mosq, on_disconnect);
    mosquitto_message_callback_set(client->mosq, on_message);
    mosquitto_publish_callback_set(client->mosq, on_publish);
    mosquitto_subscribe_callback_set(client->mosq, on_subscribe);
    mosquitto_unsubscribe_callback_set(client->mosq, on_unsubscribe);
    
    g_client = client;
    mqtt_log(3, "MQTT client initialized successfully\n");
    
    return 0;
}

// 连接到MQTT代理
int mqtt_client_connect(mqtt_client_t *client, const char* host, int port) {
    if (!client || !client->mosq) {
        mqtt_log(1, "Invalid client for connection\n");
        return -1;
    }
    
    if (host) {
        strncpy(client->broker_host, host, sizeof(client->broker_host) - 1);
    }
    if (port > 0) {
        client->broker_port = port;
    }
    
    pthread_mutex_lock(&client->mutex);
    client->state = MQTT_STATE_CONNECTING;
    pthread_mutex_unlock(&client->mutex);
    
    mqtt_log(3, "Connecting to MQTT broker %s:%d...\n", client->broker_host, client->broker_port);
    mqtt_log(3, "Client ID: %s\n", client->client_id);
    
    int ret = mosquitto_connect(client->mosq, client->broker_host, client->broker_port, MQTT_KEEPALIVE);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to connect to MQTT broker: %s\n", mosquitto_strerror(ret));
        mqtt_log(1, "Error code: %d\n", ret);
        pthread_mutex_lock(&client->mutex);
        client->state = MQTT_STATE_ERROR;
        pthread_mutex_unlock(&client->mutex);
        return -1;
    }
    
    mqtt_log(3, "Connection request sent successfully\n");
    return 0;
}

// 断开连接
int mqtt_client_disconnect(mqtt_client_t *client) {
    if (!client || !client->mosq) {
        return -1;
    }
    
    mqtt_log(3, "Disconnecting from MQTT broker...\n");
    
    int ret = mosquitto_disconnect(client->mosq);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to disconnect: %s\n", mosquitto_strerror(ret));
        return -1;
    }
    
    return 0;
}

// 清理资源
int mqtt_client_cleanup(mqtt_client_t *client) {
    if (!client) {
        return -1;
    }
    
    if (client->mosq) {
        mosquitto_loop_stop(client->mosq, true);
        mosquitto_destroy(client->mosq);
        client->mosq = NULL;
    }
    
    mosquitto_lib_cleanup();
    pthread_mutex_destroy(&client->mutex);
    
    mqtt_log(3, "MQTT client cleaned up\n");
    
    return 0;
}

// 订阅主题
int mqtt_client_subscribe(mqtt_client_t *client, const char* topic, int qos) {
    if (!client || !client->mosq || !topic) {
        mqtt_log(1, "Invalid parameters for subscribe\n");
        return -1;
    }
    
    if (client->state != MQTT_STATE_CONNECTED) {
        mqtt_log(1, "Client not connected for subscribe\n");
        return -1;
    }
    
    mqtt_log(3, "Subscribing to topic: %s (QoS %d)\n", topic, qos);
    
    int ret = mosquitto_subscribe(client->mosq, NULL, topic, qos);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to subscribe to topic '%s': %s\n", topic, mosquitto_strerror(ret));
        return -1;
    }
    
    return 0;
}

// 取消订阅主题
int mqtt_client_unsubscribe(mqtt_client_t *client, const char* topic) {
    if (!client || !client->mosq || !topic) {
        mqtt_log(1, "Invalid parameters for unsubscribe\n");
        return -1;
    }
    
    if (client->state != MQTT_STATE_CONNECTED) {
        mqtt_log(1, "Client not connected for unsubscribe\n");
        return -1;
    }
    
    mqtt_log(3, "Unsubscribing from topic: %s\n", topic);
    
    int ret = mosquitto_unsubscribe(client->mosq, NULL, topic);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to unsubscribe from topic '%s': %s\n", topic, mosquitto_strerror(ret));
        return -1;
    }
    
    return 0;
}

// 发布消息
int mqtt_client_publish(mqtt_client_t *client, const char* topic, const char* payload, int qos, int retain) {
    if (!client || !client->mosq || !topic || !payload) {
        mqtt_log(1, "Invalid parameters for publish\n");
        return -1;
    }
    
    if (client->state != MQTT_STATE_CONNECTED) {
        mqtt_log(1, "Client not connected for publish\n");
        return -1;
    }
    
    mqtt_log(3, "Publishing to topic '%s': %s\n", topic, payload);
    
    int ret = mosquitto_publish(client->mosq, NULL, topic, strlen(payload), payload, qos, retain);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to publish message: %s\n", mosquitto_strerror(ret));
        return -1;
    }
    
    return 0;
}

// 设置认证信息
int mqtt_client_set_auth(mqtt_client_t *client, const char* username, const char* password) {
    if (!client || !client->mosq) {
        return -1;
    }
    
    if (username) {
        strncpy(client->username, username, sizeof(client->username) - 1);
    }
    if (password) {
        strncpy(client->password, password, sizeof(client->password) - 1);
    }
    
    int ret = mosquitto_username_pw_set(client->mosq, 
                                       username ? client->username : NULL,
                                       password ? client->password : NULL);
    if (ret != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to set authentication: %s\n", mosquitto_strerror(ret));
        return -1;
    }
    
    mqtt_log(3, "Authentication set successfully\n");
    return 0;
}

// 设置回调函数
int mqtt_client_set_callbacks(mqtt_client_t *client, 
                             mqtt_message_callback_t msg_cb,
                             mqtt_connect_callback_t conn_cb,
                             mqtt_disconnect_callback_t disconn_cb) {
    if (!client) {
        return -1;
    }
    
    g_message_callback = msg_cb;
    g_connect_callback = conn_cb;
    g_disconnect_callback = disconn_cb;
    
    return 0;
}

// 主循环
void mqtt_client_loop(mqtt_client_t *client) {
    if (!client || !client->mosq) {
        return;
    }
    
    mqtt_log(3, "Starting MQTT client main loop...\n");
    
    time_t last_heartbeat = time(NULL);
    
    while (g_running) {
        time_t now = time(NULL);
        
        // 显示当前连接状态
        static mqtt_state_t last_state = MQTT_STATE_DISCONNECTED;
        if (client->state != last_state) {
            mqtt_log(3, "State changed: %s -> %s\n", 
                    mqtt_state_to_string(last_state), 
                    mqtt_state_to_string(client->state));
            last_state = client->state;
        }
        
        // 检查连接状态
        if (client->state == MQTT_STATE_RECONNECTING) {
            if (client->reconnect_count < MQTT_MAX_RECONNECT) {
                mqtt_log(2, "Attempting to reconnect... (attempt %d/%d)\n", 
                        client->reconnect_count + 1, MQTT_MAX_RECONNECT);
                
                if (mqtt_client_connect(client, NULL, 0) == 0) {
                    client->reconnect_count++;
                }
                sleep(MQTT_RECONNECT_DELAY);
            } else {
                mqtt_log(1, "Maximum reconnection attempts reached\n");
                break;
            }
        }
        
        // 发送心跳
        if (client->state == MQTT_STATE_CONNECTED && 
            (now - last_heartbeat) >= 30) { // 每30秒发送一次心跳
            
            char heartbeat_msg[128];
            snprintf(heartbeat_msg, sizeof(heartbeat_msg), 
                    "{\"timestamp\":%ld,\"uptime\":%ld}", now, now - client->connect_count);
            
            mqtt_client_publish(client, MQTT_TOPIC_HEARTBEAT, heartbeat_msg, MQTT_QOS_0, 0);
            last_heartbeat = now;
        }
        
        sleep(1);
    }
    
    mqtt_log(3, "MQTT client main loop stopped\n");
}

// 打印统计信息
void mqtt_client_print_stats(mqtt_client_t *client) {
    if (!client) {
        return;
    }
    
    pthread_mutex_lock(&client->mutex);
    printf("\n=== MQTT Client Statistics ===\n");
    printf("State: %s\n", mqtt_state_to_string(client->state));
    printf("Broker: %s:%d\n", client->broker_host, client->broker_port);
    printf("Client ID: %s\n", client->client_id);
    printf("Connect Count: %lu\n", client->connect_count);
    printf("Messages Sent: %lu\n", client->messages_sent);
    printf("Messages Received: %lu\n", client->messages_received);
    printf("Reconnect Count: %d\n", client->reconnect_count);
    printf("=============================\n\n");
    pthread_mutex_unlock(&client->mutex);
}

// 工具函数
const char* mqtt_state_to_string(mqtt_state_t state) {
    switch (state) {
        case MQTT_STATE_DISCONNECTED: return "DISCONNECTED";
        case MQTT_STATE_CONNECTING: return "CONNECTING";
        case MQTT_STATE_CONNECTED: return "CONNECTED";
        case MQTT_STATE_RECONNECTING: return "RECONNECTING";
        case MQTT_STATE_ERROR: return "ERROR";
        default: return "UNKNOWN";
    }
}

// 示例消息处理函数
void example_message_handler(const char* topic, const char* payload, int payload_len) {
    mqtt_log(3, "Handling message on topic '%s'\n", topic);
    
    if (strcmp(topic, MQTT_TOPIC_COMMAND) == 0) {
        // 处理设备命令
        if (strncmp(payload, "status", 6) == 0) {
            // 发布设备状态
            char status_msg[256];
            snprintf(status_msg, sizeof(status_msg), 
                    "{\"status\":\"online\",\"timestamp\":%ld}", time(NULL));
            mqtt_client_publish(g_client, MQTT_TOPIC_STATUS, status_msg, MQTT_QOS_1, 0);
        }
        else if (strncmp(payload, "restart", 7) == 0) {
            mqtt_log(2, "Restart command received\n");
            // 在实际嵌入式系统中，这里可以触发重启
        }
    }
}

void example_connect_handler(int result) {
    if (result == 0) {
        mqtt_log(3, "Application: Connected to broker\n");
        
        // 连接成功后发布在线状态
        char online_msg[128];
        snprintf(online_msg, sizeof(online_msg), 
                "{\"status\":\"online\",\"timestamp\":%ld}", time(NULL));
        mqtt_client_publish(g_client, MQTT_TOPIC_STATUS, online_msg, MQTT_QOS_1, 1);
    } else {
        mqtt_log(1, "Application: Failed to connect to broker\n");
    }
}

void example_disconnect_handler(int result) {
    mqtt_log(2, "Application: Disconnected from broker (result=%d)\n", result);
}

// 主函数
int main(int argc, char *argv[]) {
    mqtt_client_t client;
    
    // 设置信号处理
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
    
    mqtt_log(3, "Starting MQTT client application...\n");
    
    // 初始化客户端
    if (mqtt_client_init(&client, MQTT_CLIENT_ID) != 0) {
        mqtt_log(1, "Failed to initialize MQTT client\n");
        return 1;
    }
    
    // 设置回调函数
    mqtt_client_set_callbacks(&client, 
                             example_message_handler,
                             example_connect_handler,
                             example_disconnect_handler);
    
    // 设置认证信息(如果需要)
    if (MQTT_USERNAME && MQTT_PASSWORD) {
        mqtt_client_set_auth(&client, MQTT_USERNAME, MQTT_PASSWORD);
    }
    
    // 连接到代理
    if (mqtt_client_connect(&client, MQTT_BROKER_HOST, MQTT_BROKER_PORT) != 0) {
        mqtt_log(1, "Failed to initiate connection\n");
        mqtt_client_cleanup(&client);
        return 1;
    }
    
    // 启动网络循环 - 在连接请求之后启动
    mqtt_log(3, "Starting network loop...\n");
    int loop_rc = mosquitto_loop_start(client.mosq);
    if (loop_rc != MOSQ_ERR_SUCCESS) {
        mqtt_log(1, "Failed to start mosquitto loop: %s\n", mosquitto_strerror(loop_rc));
        mqtt_client_cleanup(&client);
        return 1;
    }
    
    // 等待连接建立
    int timeout = 30;  // 延长至30秒
    mqtt_log(3, "Waiting for connection to establish (timeout: %d seconds)...\n", timeout);
    while (client.state != MQTT_STATE_CONNECTED && client.state != MQTT_STATE_ERROR && timeout > 0) {
        mqtt_log(3, "Waiting... state=%s, timeout=%d\n", mqtt_state_to_string(client.state), timeout);
        sleep(1);
        timeout--;
    }
    
    if (client.state != MQTT_STATE_CONNECTED) {
        mqtt_log(1, "Connection timeout\n");
        mqtt_client_cleanup(&client);
        return 1;
    }
    
    // 订阅主题
    mqtt_client_subscribe(&client, MQTT_TOPIC_COMMAND, MQTT_QOS_1);
    
    // 发布初始状态
    char init_msg[256];
    snprintf(init_msg, sizeof(init_msg), 
            "{\"status\":\"initialized\",\"version\":\"1.0\",\"timestamp\":%ld}", 
            time(NULL));
    mqtt_client_publish(&client, MQTT_TOPIC_DATA, init_msg, MQTT_QOS_1, 0);
    
    // 主循环
    mqtt_client_loop(&client);
    
    // 停止网络循环
    mosquitto_loop_stop(client.mosq, true);
    
    // 发布下线消息
    char offline_msg[128];
    snprintf(offline_msg, sizeof(offline_msg), 
            "{\"status\":\"offline\",\"timestamp\":%ld}", time(NULL));
    mqtt_client_publish(&client, MQTT_TOPIC_STATUS, offline_msg, MQTT_QOS_1, 1);
    
    // 清理资源
    mqtt_client_print_stats(&client);
    mqtt_client_disconnect(&client);
    mqtt_client_cleanup(&client);
    
    mqtt_log(3, "MQTT client application terminated\n");
    return 0;
}

/*
编译说明:
1. 安装依赖库:
   sudo apt-get install libmosquitto-dev
   
2. 编译:
   gcc -o mqtt_client mqtt_client.c -lmosquitto -lpthread
   
3. 嵌入式交叉编译:
   arm-linux-gnueabihf-gcc -o mqtt_client mqtt_client.c -lmosquitto -lpthread
   
4. 运行:
   ./mqtt_client
   
5. 测试命令:
   mosquitto_pub -h broker.hivemq.com -t "device/command" -m "status"
   mosquitto_sub -h broker.hivemq.com -t "device/+"
*/